package com.shujia.flink.window

import java.lang

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object Demo5Car {

  def main(args: Array[String]): Unit = {

    /**
      * 实时读取卡口过车数据-- 实时统计道路拥堵情况
      * 拥堵判断条件
      * 1、最近一段时间的平均车速
      * 2、最近一段时间车的数量
      *
      * 计算最近10分钟的数据，每隔1分钟计算一次
      *
      */

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


    //读取卡口过车数据
    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)


    /**
      * 解析json数据
      *
      */
    val carDS: DataStream[(String, Long, Double)] = linesDS.map(line => {
      val jsonObj: JSONObject = JSON.parseObject(line)

      //直接通过key获取value
      val card: String = jsonObj.getString("card")
      val time: Long = jsonObj.getLong("time")
      val speed: Double = jsonObj.getDouble("speed")

      (card, time * 1000, speed)
    })

    /**
      * 设置时间字段和水位线
      *
      */

    val assDS: DataStream[(String, Long, Double)] = carDS.assignTimestampsAndWatermarks(
      //执行水位线前移的时间
      new BoundedOutOfOrdernessTimestampExtractor[(String, Long, Double)](Time.seconds(5)) {
        //指定时间戳字段, 指定的时间字段必须是毫秒级别
        override def extractTimestamp(element: (String, Long, Double)): Long = element._2
      }
    )

    /**
      *
      * 计算最近10分钟的数据，每隔1分钟计算一次
      */

    val windowDS: WindowedStream[(String, Long, Double), String, TimeWindow] = assDS
      //按照卡口分组
      .keyBy(_._1)
      .timeWindow(Time.minutes(10), Time.minutes(1))


    /**
      * 1、最近一段时间的平均车速
      * 2、最近一段时间车的数量
      *
      * 输出结果
      * 卡口，窗口的结束时间，平均车速，车的数量
      */

    val resultDS: DataStream[(String, Long, Double, Long)] = windowDS.process(new ProcessWindowFunction[(String, Long, Double), (String, Long, Double, Long), String, TimeWindow] {
      override def process(key: String,
                           context: Context,
                           elements: Iterable[(String, Long, Double)],
                           out: Collector[(String, Long, Double, Long)]): Unit = {

        var num = 0
        var sumSpeed = 0.0

        for ((card, time, speed) <- elements) {
          //统计车辆数量
          num += 1
          //总的车速
          sumSpeed += speed
        }

        //计算平均车速
        val avgSpeed: Double = sumSpeed / num

        //获取窗口的结束时间
        val endTIme: Long = context.window.getEnd

        //将数据发送到下游
        out.collect((key, endTIme, avgSpeed, num))
      }
    })

    resultDS.print()

    env.execute()


  }

}
